package ds_industry_2025.ds.ds_07.T3

import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions._

/*
    15、根据dwd或者dws层的数据，请计算每个省份累计订单量，然后根据每个省份订单量从高到低排列，将结果打印到控制台（
    使用spark中的show算子，同时需要显示列名）；
例如：可以考虑首先生成类似的临时表A：
province_name	Amount（订单量）
A省	10122
B省	301
C省	2333333

然后生成结果类似如下：其中C省销量最高，排在第一列，A省次之，以此类推。

C省	A省	B省
23333331	10122	301

 */

object t4 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("t4")
      .config("org.apache.spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()
    import spark.implicits._

    spark.table("dwd.fact_order_info")
      .where("etl_date=(select max(etl_date) from dwd.fact_order_info)")
      .createOrReplaceTempView("order_info")

    spark.table("dwd.dim_province")
      .where("etl_date=(select max(etl_date) from dwd.dim_province)")
      .createOrReplaceTempView("province_info")

    val r1 = spark.sql(
      """
        |select distinct
        |p.name as province_name,
        |count(*) over(partition by p.region_id,o.province_id,p.name) as total_count
        |from order_info as o
        |join province_info as p
        |on p.id=o.province_id
        |""".stripMargin).orderBy(desc("total_count"))


    r1.show

    //  todo 提取排好顺序的字段,后续提供给pivot使用,这样显示的提供字段的顺序可以保证结果是根据省份的订单数量排序的
    val names:Array[String] = r1.select("province_name").as[String].collect()
    /*
        为什么不需要 groupBy：
    如果你在 groupBy 中指定了分组字段（如 groupBy("province_name")），那么结果会是每个省份单独一行，而不是将所有省份的值放在同一行。

    你希望的结果是一个宽表（一行数据，多个字段），因此不需要对数据进行分组。

    在你的数据中，每个省份的 total_count 是唯一的（即每个省份只有一条记录），因此使用 sum 不会改变数据本身。
     */
    r1.groupBy()
      .pivot("province_name",names)
      .sum("total_count")
      .show(false)



    spark.close()

  }

}
